package com.shengzai.flink.tf

import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.scala._

object demo4KeyBy {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(2)

    //读取数据
    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")

    /**
     * keyBy: 将相同的数据发送到同一个下游的task中
     * key之后一班需要接一个聚合函数
     */
    val scalaDS: KeyedStream[String, String] = studentDS.keyBy(stu => {
      val sex: String = stu.split(",")(3)
      sex
    })
    // scalaDS.print()

    //java api
    val javaDS: KeyedStream[String, String] = studentDS
      .keyBy(new KeySelector[String, String] {
        override def getKey(stu: String): String = {
          val sex: String = stu.split(",")(3)
          sex
        }
      })

    javaDS.print()

    env.execute()
  }

}
